package com.doit.day03;

import com.alibaba.fastjson.JSON;
import lombok.*;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CreateData2Kafka {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.ACKS_CONFIG,"1");
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        EventLog eventLog = new EventLog();
        while (true){
            eventLog.setEventId(RandomStringUtils.randomAlphanumeric(5));
            eventLog.setGuid(RandomUtils.nextInt(100,1000000));
            eventLog.setTimestamp(System.currentTimeMillis());
            String json = JSON.toJSONString(eventLog);
            System.out.println(json);
            ProducerRecord<String, String> record = new ProducerRecord<>("event-log", json);
            producer.send(record);
            producer.flush();
            Thread.sleep(RandomUtils.nextInt(100,300));
        }
    }
}


@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@ToString
class EventLog{
    //{"guid":1,"eventId":"pageview","timestamp":1637868346789}
    private long guid;
    private String eventId;
    private long timestamp;
    private int isNew = 0;
}